package net.bwie.douyin.dwd.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.douyin.dwd.log.Function.*;
import net.bwie.douyin.dwd.log.bean.LiveEvent;
import net.bwie.realtime.douyin.common.utils.JdbcUtil;
import net.bwie.realtime.douyin.common.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.time.Duration;


public class DouYinJob {


    public static void main(String[] args) throws Exception {
        //1.执行环境-evn
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        //2.数据源-source
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "live_events");
//        kafkaDataStream.print("kafka");


        //3-数据清洗
        SingleOutputStreamOperator<String> etlDataStream = logCleaned(kafkaDataStream);


        // 4.实体类
        SingleOutputStreamOperator<LiveEvent> LiveEventBeanStream = etlDataStream.process(new ProcessFunction<String, LiveEvent>() {
            @Override
            public void processElement(String value, ProcessFunction<String, LiveEvent>.Context ctx, Collector<LiveEvent> out) throws Exception {
                LiveEvent liveEvent = JSON.parseObject(value, LiveEvent.class);
                out.collect(liveEvent);
            }
        });

//        LiveEventBeanStream.print("LiveEventBeanStream");


        //5-水位线
        SingleOutputStreamOperator<LiveEvent> timeStream = LiveEventBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<LiveEvent>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<LiveEvent>() {
                                    @Override
                                    public long extractTimestamp(LiveEvent element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

//       6-1实时在线人数统计
        DataStream<String> UvCountStream = UvCount(timeStream);
//        UvCountStream.print();





        //6-2礼物收入实时统计
        DataStream<String> doubleDataStream = GiftSum(timeStream);
//        doubleDataStream.print();


        //6-3 互动消息热度分析
        DataStream<String> hotDataStream = HotData(timeStream);
//        hotDataStream.print();


        //6-4刷礼物异常检测
        DataStream<String> GiftAbnormalStream = GiftAbnormal(timeStream);
//        GiftAbnormalStream.print();

        //6-5机器人发言检测
        DataStream<String> robotSpeakStream = robotSpeak(timeStream);
//        robotSpeakStream.print();


        //7-数据sinkToClickhouse
        //礼物收入实时统计
        JdbcUtil.sinkToClickhouseUpsert(UvCountStream, "INSERT INTO douyin_live_events.dws_douyin_UvCount_report(\n" +
                "    window_start_time, window_end_time,\n" +
                " uv_count, ts\n" +
                ")\n" +
                "VALUES (?, ?, ?, ?)");


//        礼物收入实时统计
        JdbcUtil.sinkToClickhouseUpsert(doubleDataStream, "INSERT INTO douyin_live_events.dws_douyin_RoomId_giftSum_report(\n" +
                "    window_start_time, window_end_time,\n" +
                "    getRoomId, giftSum, ts\n" +
                ")\n" +
                "VALUES (?, ?, ?, ?, ?)");



        //互动消息热度分析
        JdbcUtil.sinkToClickhouseUpsert(hotDataStream, "INSERT INTO douyin_live_events.dws_douyin_eventType_count_report(\n" +
                "    window_start_time, window_end_time,\n" +
                "    eventType, event_count, ts\n" +
                ")\n" +
                "VALUES (?, ?, ?, ?, ?)");

        //刷礼物异常检测
        JdbcUtil.sinkToClickhouseUpsert(GiftAbnormalStream, "INSERT INTO douyin_live_events.dws_douyin_GiftAbnormal_report(\n" +
                "    window_start_time, window_end_time,\n" +
                "    UserId, GiftValue_sum, ts\n" +
                ")\n" +
                "VALUES (?, ?, ?, ?, ?);");


        //机器人发言检测
        JdbcUtil.sinkToClickhouseUpsert(robotSpeakStream, "INSERT INTO douyin_live_events.dws_douyin_robotSpeak_report(\n" +
                "    window_start_time, window_end_time,\n" +
                "    UserId, count, ts\n" +
                ")\n" +
                "VALUES (?, ?, ?, ?, ?)");


        //7.执行-execute
        env.execute("JtpLogEtlJob");
    }

    private static DataStream<String> robotSpeak(SingleOutputStreamOperator<LiveEvent> timeStream) {
//        机器人发言检测 (e)
//        实现思路：
//        检测同一用户在1s内发言次数超过2次。
        SingleOutputStreamOperator<String> apply = timeStream.filter(event -> "message".equals(event.getEventType()))
                .keyBy(new KeySelector<LiveEvent, String>() {
                    @Override
                    public String getKey(LiveEvent value) throws Exception {
                        return value.getUserId();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .apply(new robotSpeakFunction());

        return apply;
    }

    private static DataStream<String> GiftAbnormal(SingleOutputStreamOperator<LiveEvent> timeStream) {
//        刷礼物异常检测 (d)
//        实现思路：
//        检测同一用户在10s内送礼超过1000。
//        使用 窗口函数聚合。
        SingleOutputStreamOperator<String> apply = timeStream.filter(event -> "gift".equals(event.getEventType()))
                .keyBy(new KeySelector<LiveEvent, String>() {
                    @Override
                    public String getKey(LiveEvent value) throws Exception {
                        return value.getUserId();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .apply(new GiftAbnormalFunction());

        return apply;


    }

    private static DataStream<String> HotData(SingleOutputStreamOperator<LiveEvent> timeStream) {
//        互动消息热度分析 (c)
//        实现思路：
//        分类消息类型（点赞、评论、分享），统计各类型消息量。
//        使用 Flink ML 实时聚类 分析热度趋势。
        SingleOutputStreamOperator<String> apply = timeStream.filter(event ->
                        "message".equals(event.getEventType()) ||
                                "share".equals(event.getEventType()) ||
                                "like".equals(event.getEventType()))
                .keyBy(o -> o.getEventType())
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
                .apply(new HotDataFunction());

        return apply;
    }

    private static DataStream<String> GiftSum(SingleOutputStreamOperator<LiveEvent> timeStream) {
//        礼物收入实时统计 (b)
//        实现思路：
//        过滤礼物事件(gift)，按时间窗口累加礼物金额。
        SingleOutputStreamOperator<String> apply = timeStream.filter(event -> "gift".equals(event.getEventType()))
                .keyBy(event -> event.getRoomId())
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
                .apply(new GiftSumFunction());
        return apply;
    }


    private static DataStream<String> UvCount(SingleOutputStreamOperator<LiveEvent> timeStream) {

        //实时在线人数统计 (a)
        //实现思路：
        //处理用户进入(enter)/离开(exit)事件，维护当前在线用户集合。
        //使用 滑动窗口 统计最近N分钟的活跃用户数。
        KeyedStream<LiveEvent, String> KeyedStream = timeStream.keyBy(o->"all");
        WindowedStream<LiveEvent, String, TimeWindow> window = KeyedStream.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)));
        SingleOutputStreamOperator<String> apply = window.apply(new OnlineUsersFunction());
        return apply;

    }
    /*
    脏数据处理
     */
    private static SingleOutputStreamOperator<String> logCleaned(DataStream<String> kafkaDataStream) {

        //a-脏数据的侧边流输出时的标记
        final OutputTag<String> dirtyTag = new OutputTag<String>("dirty-log"){};

        //b-数据清洗处理
        SingleOutputStreamOperator<String> cleanedStream = kafkaDataStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                try {
                    //a 解析json数据
                    JSON.parseObject(value);
                    //b.没有异常，解析正确，正常输出
                    out.collect(value);
                } catch (Exception e) {
                    //c.有异常，输出到侧边流
                    ctx.output(dirtyTag, value);
                }
            }
        });

        //c-测流输出
        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(dirtyStream, "douyin-dirty-log");

        //d-返回正常数据
        return cleanedStream;


    }



}
